package org.example.operator.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * rdd的算子，也就是rdd的方法
 *
 * distinct 去重
 *
 */
object Spark10_RDD_Operator_Coalesce {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val context = new SparkContext(conf)

    val rdd = context.makeRDD(List(1,2,3,4,5,6),3)
    // 缩减为2个分区
    var coaRdd: RDD[Int] = rdd.coalesce(2)

    /**
      * 需要说明的是，默认coalesce是不会将数据打乱的,如果上述条件
      * 1 2
      *
      * 3 4
      *
      * 5 6
      *
      * 缩减到2个分区的时候，会变成
      *
      * 1 2
      *
      * 3 4 5 6
      *
      * 他是将一整个分区放到另一个分区，达到缩减效果。
      *
      * 如果你希望达到均衡，他还有第二个参数
      *
      */
    coaRdd = rdd.coalesce(2,true)
    /**
      * 根据数据量缩减分区，用于大数据集过滤后，提高小数据集的执行效率
      * 在spark程序中，存在过多的小任务的时候，可以通过coalesce方法，
      * 收缩合并分区，减少分区的个数，减少任务调度成本
      *
      * 1000    1000
      *
      * 10      10
      *
      * 其实可以合并成一个分区，让一个excutor 执行，而不是两个
      */

    /**
      * 扩大分区
      *
      * 原先是2个分区，改成3个分区
      *
      * coalesce 是支持扩容的，但是前提是一定要用shuffle来进行操作的
      * 不然扩大是没有意义的
      * 但是需要实现扩容的操作，还有更快捷的方法
      *
      * 实际，底层也是用coalesce
      */
      rdd.repartition(3)

    context.stop();
  }
}
